package com.vhall.component.plugin.common.utils;

import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisURI;
import io.lettuce.core.XGroupCreateArgs;
import io.lettuce.core.XReadArgs;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.sync.RedisCommands;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

@Component
public class RedisUtil<T> {
    private static final Logger log = LoggerFactory.getLogger(RedisUtil.class);

    /**
     * 锁有效时间
     */
    private static final long LOCK_TIME = 5000L;
    @Resource
    private RedisTemplate<String, T> redisTemplate;

    /**
     * 获取缓存，null时执行dbReader并放入缓存
     *
     * @param key      缓存key
     * @param time     缓存失效时间
     * @param timeUnit 失效时间单位
     * @param dbReader 获取缓存失败时执行并放入缓存
     * @return
     */
    public T getCache(String key, Long time, TimeUnit timeUnit, DBReader<T> dbReader) {
        T hashCache = redisTemplate.opsForValue().get(key);
        if (hashCache != null) {
            return hashCache;
        }
        String lockValue = UUID.randomUUID().toString();
        boolean lockSuccess = false;
        try {
            for (; ; ) {
                if (lockSuccess = RedisLocker.tryLock(key, lockValue, LOCK_TIME)) {
                    T fromDb = dbReader.getFromDb();
                    if (fromDb == null) {
                        return null;
                    }
                    redisTemplate.opsForValue().set(key, fromDb, time, timeUnit);
                    return fromDb;
                } else {
                    try {
                        Thread.sleep(20L);
                    } catch (InterruptedException e) {
                        log.error("Interrupted!", e);
                        Thread.currentThread().interrupt();
                    }
                    hashCache = redisTemplate.opsForValue().get(key);
                    if (hashCache != null) {
                        return hashCache;
                    }
                }
            }
        } finally {
            if (lockSuccess) {
                RedisLocker.unlock(key, lockValue);
            }
        }
    }

    /**
     * 获取hash缓存，null时执行dbReader并放入缓存
     *
     * @param key      缓存key
     * @param hash     hash键
     * @param time     缓存失效时间
     * @param timeUnit 失效时间单位
     * @param dbReader 获取缓存失败时执行并放入缓存
     * @return
     */
    public T getHashCache(String key, String hash, Long time, TimeUnit timeUnit, DBReader<T> dbReader) {
        T hashCache = redisTemplate.<String, T>opsForHash().get(key, hash);
        if (hashCache != null) {
            return hashCache;
        }
        String lockValue = UUID.randomUUID().toString();
        boolean lockSuccess = false;
        try {
            for (; ; ) {
                if (lockSuccess = RedisLocker.tryLock(key, lockValue, LOCK_TIME)) {
                    T fromDb = dbReader.getFromDb();
                    if (fromDb == null) {
                        return null;
                    }
                    redisTemplate.opsForHash().put(key, hash, fromDb);
                    redisTemplate.expire(key, time, timeUnit);
                    return fromDb;
                } else {
                    try {
                        Thread.sleep(20L);
                    } catch (InterruptedException e) {
                        log.error("Interrupted!", e);
                        Thread.currentThread().interrupt();
                    }
                    hashCache = redisTemplate.<String, T>opsForHash().get(key, hash);
                    if (hashCache != null) {
                        return hashCache;
                    }
                }
            }
        } finally {
            if (lockSuccess) {
                RedisLocker.unlock(key, lockValue);
            }
        }
    }

    /**
     * 获取hash缓存，null时执行dbReader并放入缓存
     *
     * @param key      缓存key
     * @param time     缓存失效时间
     * @param timeUnit 失效时间单位
     * @param dbReader 获取缓存失败时执行并放入缓存
     * @return
     */
    public Map<String, T> getAllHash(String key, Long time, TimeUnit timeUnit, DBReader<Map<String, T>> dbReader) {
        Map<String, T> hashAllCache = redisTemplate.<String, T>opsForHash().entries(key);
        if (hashAllCache.size() > 0) {
            return hashAllCache;
        }
        String lockValue = UUID.randomUUID().toString();
        boolean lockSuccess = false;
        try {
            for (; ; ) {
                if (lockSuccess = RedisLocker.tryLock(key, lockValue, LOCK_TIME)) {
                    Map<String, T> fromDb = dbReader.getFromDb();
                    if (fromDb == null || fromDb.size() == 0) {
                        return null;
                    }
                    redisTemplate.opsForHash().putAll(key, fromDb);
                    redisTemplate.expire(key, time, timeUnit);
                    return fromDb;
                } else {
                    try {
                        Thread.sleep(20L);
                    } catch (InterruptedException e) {
                        log.error("Interrupted!", e);
                        Thread.currentThread().interrupt();
                    }
                    hashAllCache = redisTemplate.<String, T>opsForHash().entries(key);
                    if (hashAllCache.size() > 0) {
                        return hashAllCache;
                    }
                }
            }
        } finally {
            if (lockSuccess) {
                RedisLocker.unlock(key, lockValue);
            }
        }
    }

    /**
     * 获取redis连接客户端
     *
     * @param host     地址
     * @param port     端口
     * @param password 密码
     * @param database 数据库
     * @return redis客户端
     * @author yuanzh
     */
    public static RedisClient getRedisClient(String host, int port, String password, int database) {
        Duration defaultTimeout = Duration.of(10, ChronoUnit.SECONDS);
        RedisURI redisUri = RedisURI.builder().withHost(host).withPort(port).withTimeout(defaultTimeout).withDatabase(database)
                .withPassword(password).build();
        return RedisClient.create(redisUri);
    }

    /**
     * 初始化redis stream
     *
     * @param connect    客户端连接
     * @param streamName stream名称
     * @param groupName  组名称
     * @author yuanzh
     */
    public static String initStream(StatefulRedisConnection<String, String> connect, String streamName, String groupName) {
        RedisCommands<String, String> redisSyncCommands = connect.sync();
        XGroupCreateArgs args = new XGroupCreateArgs().mkstream(true);

        // 初始化stream
        Long size = redisSyncCommands.exists(streamName);
        if (size > 0) {
            log.warn("========> stream初始化结束，已存在，跳过 stream:[{}]，group:[{}]", streamName, groupName);
            return null;
        }
        // 创建分组
        //XGROUP CREATE mystream group-1 $ MKSTREAM
        return redisSyncCommands.xgroupCreate(XReadArgs.StreamOffset.from(streamName, "0"), groupName, args);
    }
}
